package com.bigdata.warehouse.dwd;


import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: com.bigdata.warehouse.dwd
 * @Author: zhangleqing
 * @CreateTime: 2025-08-22  11:37
 * @Description: TODO
 * @Version: 1.0
 */
public class DwdEntranceGuardLog {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);


        // 门禁刷卡事实表 ods_entrance_guard_log
        tEnv.executeSql(
                "CREATE TABLE ods_entrance_guard_log(\n" +
                        " id STRING,\n" +
                        " openMode SMALLINT,\n" +
                        " openTime STRING,\n" +
                        " ownerId INT,\n" +
                        " proc_time as PROCTIME()\n" +
                        ") WITH (\n" +
                        " 'connector' = 'kafka',\n" +
                        " 'topic' = 'ods_entrance_guard_log',\n" +
                        " 'properties.bootstrap.servers' = 'node101:9092',\n" +
                        " 'properties.group.id' = 'ods_entrance_guard_log',\n" +
                        " 'scan.startup.mode' = 'earliest-offset',\n" +
                        " 'format' = 'json'\n" +
                        ")"
        );


        // 用户维表 dim_base_ower
        tEnv.executeSql(
                "CREATE TABLE dim_base_ower (\n" +
                        "    id INT,\n" +
                        "    roomId INT,\n" +
                        "    owerCode STRING,\n" +
                        "    owerName STRING,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerIdcard STRING,\n" +
                        "    owerTel STRING,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    remark STRING,\n" +
                        "    createTime STRING,\n" +
                        "    operatorTime STRING\n" +
                        ") WITH (\n" +
                        "'connector' = 'doris',\n" +
                        "'fenodes' = 'node102:8030',\n" +
                        "'table.identifier' = 'sca_doris.ower',\n" +
                        "'username' = 'root',\n" +
                        "'password' = '123456'\n" +
                        ")"
        );


        // 房间维表 dim_base_room
        tEnv.executeSql(
                "CREATE TABLE dim_base_room (\n" +
                        "    id INT,\n" +
                        "    houseId INT,\n" +
                        "    roomCode STRING,\n" +
                        "    roomName STRING,\n" +
                        "    peopleNum INT,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_doris.room',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );


        // 楼栋维表 dim_base_house
        tEnv.executeSql(
                "CREATE TABLE dim_base_house (\n" +
                        "    id INT,\n" +
                        "    cid INT,\n" +
                        "    houseCode STRING,\n" +
                        "    houseName STRING,\n" +
                        "    buildStartTime STRING,\n" +
                        "    buildEndTime STRING,\n" +
                        "    houseArea STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_doris.house',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );


        // 社区维表 dim_base_community
        tEnv.executeSql(
                "CREATE TABLE dim_base_community (\n" +
                        "    id INT,\n" +
                        "    communityName STRING,\n" +
                        "    district_id INT,\n" +
                        "    city_id INT,\n" +
                        "    communityLocation STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_doris.community',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );


/* 门禁刷卡明细表
CREATE TABLE dwd_entrance_guard_log (
    id STRING,
    ownerId BIGINT,
    openTime STRING,
    cId BIGINT,
    openMode STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'dwd_entrance_guard_log',
'properties.bootstrap.servers' = 'node101:9092',
'key.format' = 'json',
'value.format' = 'json'
)
 */
        // 创建DWD层门禁刷卡明细表
        tEnv.executeSql(
                "CREATE TABLE dwd_entrance_guard_log (\n" +
                        "    id STRING,\n" +
                        "    ownerId BIGINT,\n" +
                        "    openTime STRING,\n" +
                        "    cId BIGINT,\n" +
                        "    openMode STRING,\n" +
                        "    PRIMARY KEY (id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "'connector' = 'upsert-kafka',\n" +
                        "'topic' = 'dwd_entrance_guard_log',\n" +
                        "'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "'key.format' = 'json',\n" +
                        "'value.format' = 'json'\n" +
                        ")"
        );


        tEnv.executeSql(
                "INSERT INTO dwd_entrance_guard_log \n" +
                        "SELECT \n" +
                        "    log.id, \n" +
                        "    log.ownerId, \n" +
                        "    log.openTime, \n" +
                        "    COALESCE(comm.id, -1) as cId, \n" +
                        "    CASE log.openMode \n" +
                        "        WHEN 1 THEN '刷卡开门' \n" +
                        "        WHEN 2 THEN '密码开门' \n" +
                        "        WHEN 3 THEN '指纹开门' \n" +
                        "        WHEN 4 THEN '人脸识别' \n" +
                        "        WHEN 5 THEN '远程开门' \n" +
                        "        ELSE '其他方式' \n" +
                        "    END as openMode \n" +
                        "FROM ods_entrance_guard_log log \n" +
                        "-- 关联用户维表\n" +
                        "LEFT JOIN dim_base_ower ower \n" +
                        "ON log.ownerId = ower.id \n" +
                        "-- 关联房间维表\n" +
                        "LEFT JOIN dim_base_room room \n" +
                        "ON ower.roomId = room.id \n" +
                        "-- 关联楼栋维表\n" +
                        "LEFT JOIN dim_base_house house \n" +
                        "ON room.houseId = house.id \n" +
                        "-- 关联社区维表，获取社区ID(cId)\n" +
                        "LEFT JOIN dim_base_community comm \n" +
                        "ON house.cid = comm.id"
        );

        tEnv.executeSql(
                "select * from dwd_entrance_guard_log"
        ).print();




        //env.execute("DwdEntranceGuardLog");
    }
}
